package com.guandata.cdc.model.db_type.postgres;

import com.guandata.cdc.model.source.AbstractSourceConfig;
import lombok.ToString;
import org.apache.flink.api.java.utils.ParameterTool;

import java.util.Optional;

@ToString
public class PostgresSourceConfig extends AbstractSourceConfig {
    public final String slotName;

    public final String sourceSchema;

    public final String sourceTable;

    public final PostgresSourceMode mode;

    // use pg out·put·for Postgres SQL 10+
    // or use decoder·buf·s for Postgres SQL 10- with external plugin
    public final String pluginName = "pgoutput";

    public final String connector = "postgres-cdc";

    public final PostgresStartupMode startUpMode;

    public final String sourceJdbcUrl;

    /**
     * -- experimental feature: incremental snapshot (default off)
     *
     * @return true - enable incremental snapshot
     */
    public Boolean isIncrementalMode() {
        return mode == PostgresSourceMode.INCREMENTAL;
    }

    public PostgresSourceConfig(ParameterTool parameterTool) {
        super(parameterTool);
        this.sourceSchema = parameterTool.getRequired("source.schema");
        this.sourceTable = String.format("%s.%s", sourceSchema, rawSourceTable);
        var suffix = parameterTool.get("source.slot-name.suffix", "");
        this.slotName = String.format("%s_%s%s", "flink", rawSourceTable, suffix).replaceAll("-", "_");
        this.mode = PostgresSourceMode.fromCode(parameterTool.get("source.mode"));
        this.startUpMode = Optional
            .ofNullable(parameterTool.get("source.startup.mode"))
            .map(PostgresStartupMode::fromCode)
            .orElse(PostgresStartupMode.INITIAL);
        this.sourceJdbcUrl = String.format(PostgresConst.SHORT_JDBC_URL_PREFIX, sourceHostname, sourcePort);
    }
}
